package cn.gov.cus.rmftquartz.job.entryandmanifest.entry;

import cn.gov.cus.rmftquartz.common.utils.JobUtil;
import cn.gov.cus.rmftquartz.entity.ManifestChkNewVo;
import cn.gov.cus.rmftquartz.job.JobStyle;
import cn.gov.cus.rmftquartz.listener.spring.SpringJobListener;
import cn.gov.cus.rmftquartz.thread.PrepareBatchDataTimeNoThread;
import cn.gov.cus.rmftquartz.thread.PrepareBatchDataTimeOkThread;
import cn.gov.cus.rmftquartz.service.entry.ScheduledEntryService;
import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.extern.slf4j.Slf4j;
import org.quartz.*;
import org.quartz.impl.matchers.KeyMatcher;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.scheduling.quartz.QuartzJobBean;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author wxming.
 * @date 2018/2/11.
 */
@Data
@EqualsAndHashCode(callSuper = false)
@Slf4j
@Component
@DisallowConcurrentExecution
public class ScheduledReceiverEntry extends QuartzJobBean implements JobStyle{
    private static final String JOB_GROUP = "EntryAndManifest";
    private static final String JOB_NAME = ScheduledReceiverEntry.class.getSimpleName();
    private ThreadPoolExecutor executor;


    @Value("${entryHashcode}")
    private String hashcode;
    @Value("${entryPrepareDataPoolSize}")
    private int entryPrepareDataPoolSize;
    @Value("${entryPrepareDataBatchPoolSize}")
    private int entryPrepareDataBatchPoolSize;
    /**
     * 报关单调流处理接口一批数据的最大条数
     */
    private int entryBatchNumber=30;
    /**
     * 报关单处理H2000接口表数据的时间差（分）
     */
    private long entryWaitMinutes=3;

    @Autowired
    private ScheduledEntryService scheduledEntryService;
    @Autowired
    private JobUtil jobUtil;

    /**
     * @param context JobExecutionContext
     * @see #execute
     */
    @Override
    protected void executeInternal(JobExecutionContext context) {
        try {
            List<ManifestChkNewVo> entryList = scheduledEntryService.getEntry();
            if(entryList != null && entryList.size()>0){
                executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(entryPrepareDataPoolSize);

                Date nowDate = new Date();
                List<ManifestChkNewVo> entryBatchTimeOkList = new ArrayList<>();
                List<ManifestChkNewVo> entryBatchTimeNoList = new ArrayList<>();
                int timeOkListNum=0;
                int timeNoListNum=0;
                Date batchTime = new Date();

                for (ManifestChkNewVo vo:entryList){
                    long minutes=(nowDate.getTime()-vo.getCreateDate().getTime())/(1000*60);
                    if(minutes<0){
                        //记日志。
                    }
                    //超时
                    if(minutes>entryWaitMinutes){
                        entryBatchTimeNoList.add(vo);
                        timeNoListNum++;
                        if(timeNoListNum>=entryBatchNumber){
                            timeNoListToThread(entryBatchTimeNoList);
                            entryBatchTimeNoList = new ArrayList<>();
                            timeNoListNum=0;
                        }
                    }else{
                        if(timeOkListNum==0){
                            batchTime=vo.getCreateDate();//取一批数据中最老的时间。
                        }
                        entryBatchTimeOkList.add(vo);
                        timeOkListNum++;
                        if(timeOkListNum>=entryBatchNumber){
                            timeOkListToThread(entryBatchTimeOkList,batchTime);
                            entryBatchTimeOkList = new ArrayList<>();
                            timeOkListNum=0;
                        }
                    }
                }
                if(entryBatchTimeNoList.size()>0){
                    timeNoListToThread(entryBatchTimeNoList);
                }
                if(entryBatchTimeOkList.size()>0){
                    timeOkListToThread(entryBatchTimeOkList,batchTime);
                }
                //结束线程池
                executor.shutdown();
                //等待子线程全部执行完毕（最大等待时间：1 Day)
                executor.awaitTermination(1, TimeUnit.DAYS);
            }
        }catch (Exception e){

        }
    }

    /**
     * 预防等待队列过载
     * @param poolSize
     */
    private void forExecutor(int poolSize){
        if(executor.getQueue().size() > poolSize){
            try{
                executor.awaitTermination(100,TimeUnit.MILLISECONDS);
            }catch (InterruptedException e){

            }
        }
    }

    private void timeOkListToThread(List<ManifestChkNewVo> entryBatchTimeOkList,Date batchTime){
        List<ManifestChkNewVo> entryBatchToThreadList = new ArrayList<>();
        entryBatchToThreadList.addAll(entryBatchTimeOkList);
        Runnable runnable = new PrepareBatchDataTimeOkThread(entryBatchToThreadList,batchTime,scheduledEntryService,hashcode,entryPrepareDataBatchPoolSize);
        forExecutor(entryPrepareDataPoolSize);
        executor.submit(runnable);
    }

    private void timeNoListToThread(List<ManifestChkNewVo> entryBatchTimeNoList){
        List<ManifestChkNewVo> entryBatchToThreadList = new ArrayList<>();
        entryBatchToThreadList.addAll(entryBatchTimeNoList);
        Runnable runnable = new PrepareBatchDataTimeNoThread(entryBatchToThreadList,scheduledEntryService,entryPrepareDataBatchPoolSize);
        forExecutor(entryPrepareDataPoolSize);
        executor.submit(runnable);
    }

    /**
     * @return JobDetail
     */
    @Override
    public JobDetail getJobDetail() {
        return JobBuilder.newJob(ScheduledReceiverEntry.class)
                .withIdentity(this.getClass().getSimpleName(), JOB_GROUP)
                .build();
    }

    /**
     * @return Trigger
     */
    @Override
    public Trigger getJobTrigger() {
        CronScheduleBuilder scheduleBuilder = CronScheduleBuilder.cronSchedule("0/2 * * * * ?");

        JobDataMap dataMap = new JobDataMap();

        return TriggerBuilder.newTrigger().forJob(getJobDetail())
                .withIdentity(this.getClass().getSimpleName() + "Trigger", JOB_GROUP)
                .usingJobData(dataMap)
                .startNow()
                .withSchedule(scheduleBuilder).build();
    }

    /**
     * 任务自带启动器,方便任务初始化
     * ConditionalOnProperty 注解实现任务启动的开关化.
     *
     * @param scheduler 虽然没有显示注入,但此处注入的是在 QuartzConfig 内的 quartzScheduler() 方法实例化的.
     * @return 本任务实例
     */
    @Bean(name = "ScheduledReceiverEntryExecutor")
    @ConditionalOnProperty(value = {"Quartz.Jobs.EntryAndManifest.ScheduledReceiverEntry.Enable"})
    public ScheduledReceiverEntry executeThisJob(Scheduler scheduler, SpringJobListener springJobListener) throws SchedulerException{

        JobDetail jobDetail = getJobDetail();
        Trigger jobTrigger = getJobTrigger();

        // 此处设定仅对当前 Job(JobKey 作为条件) 进行监听
        scheduler.getListenerManager().addJobListener(springJobListener, KeyMatcher.keyEquals(jobDetail.getKey()));
        jobUtil.scheduleClusterJob(scheduler,jobDetail,jobTrigger);
        return this;
    }

}
